package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.data_etl.udfs.TimeTruncUdf;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/28
 * @Desc: 学大数据，上多易教育
 *    视频播放行为，olap分析，轻度聚合表开发
 **/
public class Job4_视频行为olap分析主题轻度聚合模型 {
    public static void main(String[] args) {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        /**
         * 创建逻辑表（连接器表），映射kafka中物理topic（维度打宽明细行为日志）
         */
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                          "
                        +"     user_id           BIGINT,                     "
                        +"     username          string,                     "
                        +"     session_id        string,                     "
                        +"     event_Id          string,                     "
                        +"     event_time        bigint,                     "
                        +"     lat               double,                     "
                        +"     lng               double,                     "
                        +"     release_channel   string,                     "
                        +"     device_type       string,                     "
                        +"     properties        map<string,string>,         "
                        +"     register_phone    STRING,                     "
                        +"     user_status       INT,                        "
                        +"     register_time     TIMESTAMP(3),               "
                        +"     register_gender   INT,                        "
                        +"     register_birthday DATE,                       "
                        +"     register_city        STRING,                  "
                        +"     register_job         STRING,                  "
                        +"     register_source_type INT,                     "
                        +"     gps_province STRING,                          "
                        +"     gps_city     STRING,                          "
                        +"     gps_region   STRING,                          "
                        +"     url_prefix    STRING,                         "
                        +"     page_type    STRING,                          "
                        +"     page_service STRING,                          "
                        +"     proc_time AS proctime(),                      " // processing time 时间语义
                        +"     rt AS  to_timestamp_ltz(event_time,3),        " // 表达式字段，用于将event_time转成timestamp(3)类型
                        +"     WATERMARK FOR rt AS  rt - INTERVAL '0' SECOND " // 基于rt字段定义watermark，从此，rt字段就具备了 flink中事件时间语义
                        +" ) WITH (                                          "
                        +"  'connector' = 'kafka',                           "
                        +"  'topic' = 'dwd_events',                          "
                        +"  'properties.bootstrap.servers' = 'doitedu:9092', "
                        +"  'properties.group.id' = 'testGroup',             "
                        +"  'scan.startup.mode' = 'latest-offset',           "
                        +"  'value.format'='json',                           "
                        +"  'value.json.fail-on-missing-field'='false',      "
                        +"  'value.fields-include' = 'EXCEPT_KEY')           "
        );

        //tenv.executeSql("select * from dwd_kafka").print();

        /* *
         * 主题事实数据抽取
         */
        tenv.executeSql(
                " create temporary view video_events AS                                                 "+
                        " select                                                                                 "+
                        "   user_id,                                                                             "+
                        " 	event_Id,                                                                            "+
                        " 	device_type,                                                                         "+
                        " 	gps_province,                                                                        "+
                        " 	event_time,                                                                          "+
                        " 	rt,                                                                                  "+
                        " 	cast(properties['video_id'] as bigint) as video_id,                                   "+
                        " 	properties['play_id'] as play_id                                                     "+
                        " from dwd_kafka                                                                         "+
                        " where event_Id in ('video_play','video_hb','video_pause','video_resume','video_stop')  "
        );

        //tenv.executeSql("select * from video_events").print();
/*
+--------+--------------+------------+--------------+---------------+---------+-----------+
|user_id |     event_Id |device_type | gps_province |    event_time |video_id |   play_id |
+--------+--------------+------------+--------------+---------------+---------+-----------+
|      3 |   video_play |        mi8 |       河北省 | 1686046282002 |       1 |  play0001 |
|      3 |     video_hb |        mi8 |       河北省 | 1686046282003 |       1 |  play0001 |
|      3 |     video_hb |        mi8 |       河北省 | 1686046282004 |       1 |  play0001 |
|      3 |     video_hb |        mi8 |       河北省 | 1686046282005 |       1 |  play0001 |
|      3 |  video_pause |        mi8 |       河北省 | 1686046282006 |       1 |  play0001 |
|      3 | video_resume |        mi8 |       河北省 | 1686046282007 |       1 |  play0001 |
|      3 |     video_hb |        mi8 |       河北省 | 1686046282008 |       1 |  play0001 |
|      3 |     video_hb |        mi8 |       河北省 | 1686046282009 |       1 |  play0001 |
|      3 |     video_hb |        mi8 |       河北省 | 1686046282011 |       1 |  play0001 |
|      3 |   video_stop |        mi8 |       河北省 | 1686046282012 |       1 |  play0001 |

|      4 |   video_play |        mi8 |       河北省 | 1686046282014 |       3 |  play0002 |
|      4 |     video_hb |        mi8 |       河北省 | 1686046282015 |       3 |  play0002 |
|      4 |     video_hb |        mi8 |       河北省 | 1686046282016 |       3 |  play0002 |
|      4 |     video_hb |        mi8 |       河北省 | 1686046282017 |       3 |  play0002 |

|      4 |   video_play |        mi8 |       河北省 | 1686046282018 |       3 |  play0003 |
|      4 |     video_hb |        mi8 |       河北省 | 1686046282019 |       3 |  play0003 |
|      4 |  video_pause |        mi8 |       河北省 | 1686046282020 |       3 |  play0003 |
|      4 | video_resume |        mi8 |       河北省 | 1686046282021 |       3 |  play0003 |

|      4 |     video_hb |        mi8 |       河北省 | 1686046282022 |       4 |  play0004 |
|      4 |     video_hb |        mi8 |       河北省 | 1686046282023 |       4 |  play0004 |
|      4 |     video_hb |        mi8 |       河北省 | 1686046282024 |       4 |  play0004 |
|      4 |   video_stop |        mi8 |       河北省 | 1686046282025 |       4 |  play0004 |
 */

        /**
         * 先求出每一次视频播放行为的时长
         *    思路：遇到一个暂停，就打 标记1，否则0
         *         然后，对标记字段做 sum() over() 得到新标记
         *         然后，用新标记-老标记得到差值
         *         然后，按各种维度+差值 ，作为分组key，就可以正确分组
         */

        // 打标记，求标记差值
        tenv.executeSql(
                "create temporary view diff_tmp   AS                                                                                                                    \n" +
                        "SELECT                                                                           \n" +
                        "    user_id,                                                                     \n" +
                        "    event_Id,                                                                    \n" +
                        "    device_type,                                                                 \n" +
                        "    gps_province,                                                                \n" +
                        "    event_time,                                                                  \n" +
                        "    video_id,                                                                    \n" +
                        "    play_id,                                                                     \n" +
                        "    flag_a,                                                                      \n" +
                        "    rt,                                                                          \n" +
                        "    sum(flag_a) over(partition by play_id order by rt)  -  flag_a as diff        \n" +
                        "FROM                                                                             \n" +
                        "(                                                                                \n" +
                        "    select                                                                       \n" +
                        "        user_id,                                                                 \n" +
                        "        event_Id,                                                                \n" +
                        "        device_type,                                                             \n" +
                        "        gps_province,                                                            \n" +
                        "        event_time,                                                              \n" +
                        "        video_id,                                                                \n" +
                        "        play_id,                                                                 \n" +
                        "        rt,                                                                      \n" +
                        "        if(event_Id = 'video_pause', 1, 0 ) as flag_a                            \n" +
                        "    from video_events                                                            \n" +
                        ") o1  "
        );

        // 求“暂停”分段播放时长
        tenv.executeSql("create temporary view part_time_long as         \n" +
                "select                                                           \n" +
                "    window_start,                                                \n" +
                "    window_end,                                                  \n" +
                "    user_id,                                                     \n" +
                "    device_type,                                                 \n" +
                "    gps_province,                                                \n" +
                "    video_id,                                                    \n" +
                "    play_id,                                                     \n" +
                "    diff,                                                        \n" +
                "    max(event_time) - min(event_time) as play_time_long          \n" +
                "FROM  TABLE(                                                     \n" +
                "    TUMBLE(TABLE diff_tmp,DESCRIPTOR(rt),INTERVAL '1' MINUTE)    \n" +
                ")                                                                \n" +
                "GROUP BY                                                         \n" +
                "    window_start,                                                \n" +
                "    window_end,                                                  \n" +
                "    user_id,                                                     \n" +
                "    device_type,                                                 \n" +
                "    gps_province,                                                \n" +
                "    video_id,                                                    \n" +
                "    play_id,                                                     \n" +
                "    diff                                                        ");

        //tenv.executeSql("select * from part_time_long").print();
/*
+----+-------------------------+-------------------------+---------+--------------+----------------+------------+---------+-------------+----------------------+
| op |            window_start |              window_end | user_id |  device_type |   gps_province |   video_id | play_id |        diff |       play_time_long |
+----+-------------------------+-------------------------+---------+--------------+----------------+------------+---------+-------------+----------------------+
| +I | 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |       3 |          mi8 |         河北省 |          1 |  play0001 |           0 |                    4 |
| +I | 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |       3 |          mi8 |         河北省 |          1 |  play0001 |           1 |                    5 |

| +I | 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |       4 |          mi8 |         河北省 |          3 |  play0002 |           0 |                    3 |

| +I | 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |       4 |          mi8 |         河北省 |          3 |  play0003 |           1 |                    0 |
| +I | 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |       4 |          mi8 |         河北省 |          3 |  play0003 |           0 |                    2 |

| +I | 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |       4 |          mi8 |         河北省 |          4 |  play0004 |           0 |                    3 |
*/

        /**
         * 求一次播放的总时长： 各暂停分段的时长累加
         */
        tenv.createTemporaryFunction("time_trunc", TimeTruncUdf.class);
        tenv.executeSql("create temporary view per_play_timelong as " +
                " select                                                          " +
                "    to_date(date_format(window_end,'yyyy-MM-dd')) as dt,        \n" +
                "    time_trunc(window_end,60) as  time_60m,                     \n" +
                "    user_id,                                                    \n" +
                "    device_type,                                                \n" +
                "    gps_province,                                               \n" +
                "    video_id,                                                   \n" +
                "    play_id,                                                    \n" +
                "    proctime() as pt,                                           \n" +
                "    sum(play_time_long) as play_time_long                       \n" +
                "from part_time_long                                             \n" +
                "group by                                                        \n" +
                "    window_start,                                               \n" +
                "    window_end,                                                 \n" +
                "    user_id,                                                    \n" +
                "    device_type,                                                \n" +
                "    gps_province,                                               \n" +
                "    video_id,                                                   \n" +
                "    play_id ");

/*
+-------------------------+-------------------------+----------+--------------+--------------+----------+--------+----------------+
|            window_start |              window_end |  user_id |  device_type | gps_province | video_id |play_id | play_time_long |
+-------------------------+-------------------------+----------+--------------+--------------+----------+--------+----------------+
| 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |        3 |          mi8 |       河北省 |         1 |play0001 |   9           |
| 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |        4 |          mi8 |       河北省 |         3 |play0002 |   3           |
| 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |        4 |          mi8 |       河北省 |         3 |play0003 |   2           |
| 2023-06-06 18:11:00.000 | 2023-06-06 18:12:00.000 |        4 |          mi8 |       河北省 |         4 |play0004 |   3           |
 */

        /**
         *  创建hbase连接器表，映射hbase中的物理维表
         */
        tenv.executeSql("CREATE TABLE dim_video_info_hbase( " +
                " id BIGINT, " +
                " f ROW<video_type STRING, video_album STRING, video_author STRING,video_timelong BIGINT, create_time TIMESTAMP(3)>, " +
                " PRIMARY KEY (id) NOT ENFORCED " +
                ") WITH (                             " +
                " 'connector' = 'hbase-2.2',          " +
                " 'table-name' = 'dim_video_info',    " +
                " 'zookeeper.quorum' = 'doitedu:2181' " +
                ")");

        //tenv.executeSql("select * from  dim_video_info_hbase ").print();


        /**
         * 再将上面 “每次播放时长“结果数据 per_play_timelong ，去关联 hbase中的视频信息维表  dim_video_info_hbase
         */
        tenv.executeSql(
                        "SELECT                             "+
                        "  a.dt                                      "+
                        " ,a.time_60m                                "+
                        " ,a.user_id                                 "+
                        " ,a.device_type                             "+
                        " ,a.gps_province                            "+
                        " ,a.video_id                                "+
                        " ,a.play_id                                 "+
                        " ,b.video_album                             "+
                        " ,b.video_timelong as video_time_amt        "+
                        " ,a.play_time_long as play_time_amt         "+
                        " from per_play_timelong as a                "+
                        " left join dim_video_info_hbase             "+
                        " for system_time as of a.pt as b            "+
                        " on a.video_id = b.id                       ").print();

/*
+------------+----------------------+---------+--------------+-------------+----------+---------------+--------------+-----------------+----------------+
|         dt |             time_60m | user_id |  device_type |gps_province | video_id |       play_id |  video_album |  video_time_amt |  play_time_amt |
+------------+----------------------+---------+--------------+-------------+----------+---------------+--------------+-----------------+----------------+
| 2023-06-06 |  2023-06-06 18:00:00 |       3 |          mi8 |       河北省 |       1  |      play0001 |    咖啡与健康  |             600 |     9         |
| 2023-06-06 |  2023-06-06 18:00:00 |       4 |          mi8 |       河北省 |       3  |      play0002 |    咖啡与健康  |             500 |     3         |
| 2023-06-06 |  2023-06-06 18:00:00 |       4 |          mi8 |       河北省 |       4  |      play0004 |      硬盘选购  |            1200 |     3         |
| 2023-06-06 |  2023-06-06 18:00:00 |       4 |          mi8 |       河北省 |       3  |      play0003 |    咖啡与健康  |             500 |     2         |
*/


    }

}
